#include <glib.h>
#include <string.h>
#include <stdio.h>
#include "mqtt_task.h"
#include "log.h"
#include "timestamp.h"

#include <mosquitto.h>


gpointer mqtt_task(gpointer data);
gpointer mqtt_pub_task(gpointer data);

static GThread*     mqtt_task_h = NULL;
static GThread*     mqtt_pub_task_h = NULL;

static GAsyncQueue* mqtt_queue = NULL;

static struct mosquitto*    mqtt_client;

static int last_pack_no = -1;


#define SERVER_ADDRESS  "localhost"
// #define SERVER_ADDRESS  "192.168.2.88"
#define SERVER_PORT     1883
#define CLIENTID        "XiamenPWSAgent"
#define TOPIC           "Xiamen/PWS"
#define QOS             1
#define TIMEOUT         10000L

#define TOPIC_RF        "Xiamen/RF"

void onConnect(struct mosquitto *mosq, void *obj, int result)
{
    log_print(LOG_INFO, "MQTT server connected.");
}

void onDisconnect(struct mosquitto *mosq, void *obj, int result)
{
    log_print(LOG_ERROR, "MQTT server disconnected, reason:%d", result);
}

void onPublish(struct mosquitto *mosq, void *obj, int mid)
{

}

void mqttInit()
{
    int     rc, major, minor, rev;;

    mqtt_queue = g_async_queue_new();

    mosquitto_lib_init();

    rc = mosquitto_lib_version(&major, &minor, &rev);
    log_print(LOG_INFO, "mosquitto library: %d.%d.%d", major, minor, rev);

    mqtt_client = mosquitto_new(CLIENTID, false, NULL);

    mosquitto_connect_callback_set(mqtt_client, onConnect);
    mosquitto_publish_callback_set(mqtt_client, onPublish);
    mosquitto_disconnect_callback_set(mqtt_client, onDisconnect);

    mosquitto_connect(mqtt_client, SERVER_ADDRESS, SERVER_PORT, 60);

    mqtt_task_h = g_thread_new("MQTT thread", mqtt_task, NULL);
    mqtt_pub_task_h = g_thread_new("MQTT Pub thread", mqtt_pub_task, NULL);
}

int outDataJson(const WS_DATA_MQTT* data, char* buffer, int bufferSize)
{
    int             pos = 0;
    const WS_DATA*  d = &data->out_data;
    char            timestamp_s[30];

    if (last_pack_no == d->no) {
        return 0;
    }

    last_pack_no = d->no;

    getTimestamp(timestamp_s, sizeof(timestamp_s));
    pos = snprintf(buffer, bufferSize, "{\"t\":\"%s\",", timestamp_s);
    pos += snprintf(buffer + pos, bufferSize - pos, "\"no\":%d,", d->no);
    pos += snprintf(buffer + pos, bufferSize - pos, "\"oT\":%d.%02d,", d->temperature / 100, d->temperature % 100);
    pos += snprintf(buffer + pos, bufferSize - pos, "\"oH\":%d.%02d,", d->humidity / 100, d->humidity % 100);
    pos += snprintf(buffer + pos, bufferSize - pos, "\"wD\":%d,", d->wind_dir);
    pos += snprintf(buffer + pos, bufferSize - pos, "\"wS\":%d.%01d,", d->wind_speed / 10, d->wind_speed % 10);
    pos += snprintf(buffer + pos, bufferSize - pos, "\"wgD\":%d,", d->wind_gust_dir);
    pos += snprintf(buffer + pos, bufferSize - pos, "\"wgS\":%d.%01d,", d->wind_gust_speed / 10, d->wind_gust_speed % 10);
    pos += snprintf(buffer + pos, bufferSize - pos, "\"p\":%u.%01d,", d->pressure / 10, d->pressure % 10);
    pos += snprintf(buffer + pos, bufferSize - pos, "\"sr\":%d,", d->solar_radiation);
    pos += snprintf(buffer + pos, bufferSize - pos, "\"lux\":%u,", d->illumination);
    pos += snprintf(buffer + pos, bufferSize - pos, "\"uvi\":%d.%02d,", d->uv_index / 100, d->uv_index % 100);
    pos += snprintf(buffer + pos, bufferSize - pos, "\"uva\":%d.%02d,", d->uva_value / 100, d->uva_value % 100);
    pos += snprintf(buffer + pos, bufferSize - pos, "\"uvb\":%d.%02d,", d->uvb_value / 100, d->uvb_value % 100);
    pos += snprintf(buffer + pos, bufferSize - pos, "\"rain\":%u.%01d,", d->rain / 10, d->rain % 10);
    pos += snprintf(buffer + pos, bufferSize - pos, "\"ld\":%d,", d->lightning_dist);


    pos += snprintf(buffer + pos, bufferSize - pos, "\"oBV\":%d.%03d,", d->battery_voltage / 1000, d->battery_voltage % 1000);
    pos += snprintf(buffer + pos, bufferSize - pos, "\"oBmA\":%d.%d,", d->battery_current / 10, abs(d->battery_current) % 10);

    pos += snprintf(buffer + pos, bufferSize - pos, "\"pv\":%d.%03d}", d->solar_battery / 1000, d->solar_battery % 1000);

    return pos;
}

int RFJson(const WS_DATA_MQTT* data, char* buffer, int bufferSize)
{
    int             pos = 0;
    const WS_DATA*  d = &data->out_data;
    char            timestamp_s[30];

    getTimestamp(timestamp_s, sizeof(timestamp_s));
    pos = snprintf(buffer, bufferSize, "{\"t\":\"%s\",", timestamp_s);
    pos += snprintf(buffer + pos, bufferSize - pos, "\"no\":%d,", d->no);
    pos += snprintf(buffer + pos, bufferSize - pos, "\"rssi\":%d,", d->rssi);
    pos += snprintf(buffer + pos, bufferSize - pos, "\"snr\":%d}", d->snr);

    return pos;
}

gpointer mqtt_task(gpointer data) {
    WS_DATA_MQTT*   wsData;
    guint64         timeout;

    timeout = 500 * 1000;      // microseconds

    while(1) {
        wsData = (WS_DATA_MQTT*)g_async_queue_timeout_pop(mqtt_queue, timeout);

        if (!wsData) {
            continue;
        }

        // 格式化为JSON，发送到MQTT服务器
        char        wsDataJson[500];
        int         payloadSize = 0;
        char        wsRFJson[100];
        int         wsRFSize = 0;

        switch (wsData->type)
        {
            case TYPE_OUT:
                payloadSize = outDataJson(wsData, wsDataJson, sizeof(wsDataJson));
                wsRFSize = RFJson(wsData, wsRFJson, sizeof(wsRFJson));
                break;

            case TYPE_IN:
                break;
        }

        if (payloadSize > 0) {
            mosquitto_publish(mqtt_client, NULL, TOPIC, payloadSize, wsDataJson, 0, 0);
        }

        if (wsRFSize > 0) {
            mosquitto_publish(mqtt_client, NULL, TOPIC_RF, wsRFSize, wsRFJson, 0, 0);
        }

        free(wsData);
        wsData = NULL;
    }
    return 0;
}

void mqttPush(const WS_DATA* wsData)
{
    WS_DATA_MQTT*   data;

    data = malloc(sizeof(WS_DATA_MQTT));

    clock_gettime(CLOCK_REALTIME, &data->tm);
    data->type = TYPE_OUT;
    data->out_data = *wsData;

    g_async_queue_push(mqtt_queue, data);
}

gpointer mqtt_pub_task(gpointer data)
{
    mosquitto_loop_forever(mqtt_client, 100, 1);
}
